package org.apache.flink.runtime.resourceconsumer;

import static java.util.Objects.requireNonNull;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

import java.io.IOException;
import java.io.Serializable;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;

import javax.annotation.Nullable;

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ConsumingResult;
import org.apache.flink.api.common.ConsumingStatus;
import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.runtime.ErrorInfo;
import org.apache.flink.runtime.allocation.ResourceAllocation;
import org.apache.flink.runtime.client.ConsumeCancellationException;
import org.apache.flink.runtime.client.ConsumeException;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.SerializedThrowable;
import org.apache.flink.util.SerializedValue;

/**
 *
 * <p>This is used by the {@link ResourceConsumer} to send the results to the {@link RestEndpoint}.
 */
public class ConsumeResult implements Serializable {

    private static final long serialVersionUID = 1L;

    private final ConsumerID consumerId;

    private final ApplicationStatus applicationStatus;

    private final Map<String, SerializedValue<OptionalFailure<Object>>> accumulatorResults;

    private final long netRuntime;

    /** Stores the cause of the job failure, or {@code null} if the job finished successfully. */
    @Nullable private final SerializedThrowable serializedThrowable;

    private ConsumeResult(
            final ConsumerID consumerId,
            final ApplicationStatus applicationStatus,
            final Map<String, SerializedValue<OptionalFailure<Object>>> accumulatorResults,
            final long netRuntime,
            @Nullable final SerializedThrowable serializedThrowable) {

        checkArgument(netRuntime >= 0, "netRuntime must be greater than or equals 0");

        this.consumerId = requireNonNull(consumerId);
        this.applicationStatus = requireNonNull(applicationStatus);
        this.accumulatorResults = requireNonNull(accumulatorResults);
        this.netRuntime = netRuntime;
        this.serializedThrowable = serializedThrowable;
    }

    /** Returns {@code true} if the job finished successfully. */
    public boolean isSuccess() {
        return applicationStatus == ApplicationStatus.SUCCEEDED
                || (applicationStatus == ApplicationStatus.UNKNOWN && serializedThrowable == null);
    }

    public ConsumerID getJobId() {
        return consumerId;
    }

    public ApplicationStatus getApplicationStatus() {
        return applicationStatus;
    }

    public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorResults() {
        return accumulatorResults;
    }

    public long getNetRuntime() {
        return netRuntime;
    }

    /**
     * Returns an empty {@code Optional} if the job finished successfully, otherwise the {@code
     * Optional} will carry the failure cause.
     */
    public Optional<SerializedThrowable> getSerializedThrowable() {
        return Optional.ofNullable(serializedThrowable);
    }

    /**
     * Converts the {@link ConsumeResult} to a {@link ConsumingResult}.
     *
     * @param classLoader to use for deserialization
     * @return JobExecutionResult
     * @throws ConsumeCancellationException if the job was cancelled
     * @throws JobExecutionException if the job execution did not succeed
     * @throws IOException if the accumulator could not be deserialized
     * @throws ClassNotFoundException if the accumulator could not deserialized
     */
    public ConsumingResult toJobExecutionResult(ClassLoader classLoader)
            throws ConsumeException, IOException, ClassNotFoundException {
        if (applicationStatus == ApplicationStatus.SUCCEEDED) {
            return new ConsumingResult(
                    consumerId,
                    netRuntime,
                    AccumulatorHelper.deserializeAccumulators(accumulatorResults, classLoader));
        } else {
            final Throwable cause;

            if (serializedThrowable == null) {
                cause = null;
            } else {
                cause = serializedThrowable.deserializeError(classLoader);
            }

            final ConsumeException exception;

            if (applicationStatus == ApplicationStatus.FAILED) {
                exception = new ConsumeException(consumerId, "Job execution failed.", cause);
            } else if (applicationStatus == ApplicationStatus.CANCELED) {
                exception = new ConsumeCancellationException(consumerId, "Job was cancelled.", cause);
            } else {
                exception =
                        new ConsumeException(
                                consumerId,
                                "Job completed with illegal application status: "
                                        + applicationStatus
                                        + '.',
                                cause);
            }

            throw exception;
        }
    }

    /** Builder for {@link ConsumeResult}. */
    @Internal
    public static class Builder {

        private ConsumerID jobId;

        private ApplicationStatus applicationStatus = ApplicationStatus.UNKNOWN;

        private Map<String, SerializedValue<OptionalFailure<Object>>> accumulatorResults;

        private long netRuntime = -1;

        private SerializedThrowable serializedThrowable;

        public Builder jobId(final ConsumerID jobId) {
            this.jobId = jobId;
            return this;
        }

        public Builder applicationStatus(final ApplicationStatus applicationStatus) {
            this.applicationStatus = applicationStatus;
            return this;
        }

        public Builder accumulatorResults(
                final Map<String, SerializedValue<OptionalFailure<Object>>> accumulatorResults) {
            this.accumulatorResults = accumulatorResults;
            return this;
        }

        public Builder netRuntime(final long netRuntime) {
            this.netRuntime = netRuntime;
            return this;
        }

        public Builder serializedThrowable(final SerializedThrowable serializedThrowable) {
            this.serializedThrowable = serializedThrowable;
            return this;
        }

        public ConsumeResult build() {
            return new ConsumeResult(
                    jobId,
                    applicationStatus,
                    accumulatorResults == null ? Collections.emptyMap() : accumulatorResults,
                    netRuntime,
                    serializedThrowable);
        }
    }

   
}
